package com.bright.ghj.overt.mq.impl;

import com.bright.ghj.common.util.DateUtil;
import com.bright.ghj.overt.constant.RocketConstant;
import com.bright.ghj.overt.pojo.command.ImportAccessCommand;
import com.bright.ghj.overt.pojo.po.primary.MqMessage;
import com.bright.ghj.overt.repository.primary.MqMessageRepository;
import com.bright.ghj.overt.service.BaseDataService;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;

import java.io.File;
import java.util.Enumeration;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Tz/hxj
 * @Date 2024/2/23 10:53
 * @Description access导入队列消费类
 */
@Component
@ConditionalOnBean(value = RocketMQTemplate.class)
@RequiredArgsConstructor
@RocketMQMessageListener(topic = RocketConstant.TOPIC_CONSUMER, consumerGroup = RocketConstant.TOPIC_CONSUMER + "_group")
public class RocketConsumerServiceImpl implements RocketMQListener<MqMessage> {

    private final BaseDataService baseDataService;
    private final MqMessageRepository mqMessageRepository;
    private final ObjectMapper objectMapper;

    /**
     * 原子操作
     */
    private static ConcurrentHashMap concurrentHashMap = new ConcurrentHashMap();


    @Override
    public void onMessage(MqMessage mqMessage) {
        if (mqMessage.getId() == null) {
            System.out.println("出现空id消息，已跳过不处理");
            return;
        }

        Long startTime = System.currentTimeMillis();

        String data = (String) mqMessage.getData();
        File file = mqMessage.getFile();
        mqMessage.setContentType(1);

        //锁对象
        Object object = null;


//        synchronized (this) {


        //需要二次判断，处理锁
        synchronized (this){
            //查看这条消息是否是未消费且可用的
            MqMessage resMqMessage = mqMessageRepository.findById(mqMessage.getId()).get();

            //如果不可用 代表该消息在没被消费前被取消了
            if(!resMqMessage.getRunFlag()){
                resMqMessage.setContent("任务被取消！");
                resMqMessage.setConsumerFlag(-3);
                resMqMessage.setUpdatedBy(resMqMessage.getUsername());
                resMqMessage.setUpdatedTime(DateUtil.getCurrDate());
                mqMessageRepository.save(resMqMessage);
                return;
            }


            //如果根据key查询到了对象锁
            Enumeration enumeration = concurrentHashMap.keys();
            while(enumeration.hasMoreElements()){
                String key = (String) enumeration.nextElement();
                if(key.startsWith(mqMessage.getDistNo()) || mqMessage.getDistNo().startsWith(key)){
                    object = concurrentHashMap.get(key);
                }
            }

            if(object == null){
                object = new Object();
            }

        }



        //进入消费 表示该消息已经开始消费了

        //如果上一次执行的地区的锁没有释放, 现在有需要没释放的锁，就会进行等待
        synchronized(object) {
            try {
                startTime = System.currentTimeMillis();

                //查看这条消息是否是未消费且可用的
                MqMessage resMqMessage = mqMessageRepository.findById(mqMessage.getId()).get();
                //如果不可用 代表该消息在没被消费前被取消了
                if(!resMqMessage.getRunFlag()){
                    resMqMessage.setContent("任务被取消！");
                    resMqMessage.setConsumerFlag(-3);
                    resMqMessage.setUpdatedBy(resMqMessage.getUsername());
                    resMqMessage.setUpdatedTime(DateUtil.getCurrDate());
                    mqMessageRepository.save(resMqMessage);
                    return;
                }

                concurrentHashMap.put(mqMessage.getKeyword(), object);
                mqMessage.setOkFlag(true);
                MqMessage updResMessage = mqMessageRepository.save(mqMessage);

                if (Objects.isNull(updResMessage)) {
                    throw new RuntimeException("更新是否消费状态异常!");
                }

                if (mqMessage.getTopicType().equals(RocketConstant.TOPIC_ACCESS_IMPORT)) {

    //                Map map = objectMapper.readValue(data, Map.class);
    //                String accessFileName = (String) map.get("accessFile");

                    // 业务开始
                    try {
                        ImportAccessCommand command = objectMapper.readValue(data, ImportAccessCommand.class);
                        baseDataService.importAccess(command, file, o -> {});
                        mqMessage.setConsumerFlag(RocketConstant.CONSUMER_FLAG_SUCCESS);
                        mqMessage.setContent("导入完成！");
                    } catch (Exception e) {
                        e.printStackTrace();
                        mqMessage.setConsumerFlag(RocketConstant.CONSUMER_FLAG_BUSINESS_FAIL);
                        mqMessage.setContent("导入出错：" + e.getMessage());
                    }
                }

            } catch (Exception e){
                e.printStackTrace();
            } finally {
                //释放执行该keyword持有的锁
                concurrentHashMap.remove(mqMessage.getKeyword());
            }
        }
//        }

        Long endTime = System.currentTimeMillis();
        mqMessage.setExecTime(endTime - startTime);
        mqMessage.setUpdatedBy(mqMessage.getUsername());
        mqMessage.setUpdatedTime(DateUtil.getCurrDate());
        mqMessageRepository.save(mqMessage);

    }

}
